package com.chukun.flink.stream.operator.base;

import com.chukun.flink.stream.bean.Trade;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import scala.Tuple2;

import java.util.ArrayList;

/**
 * @author chukun
 * @version 1.0.0
 * @description reduce 操作符
 * @createTime 2022年05月10日 00:06:00
 */
public class ReduceOperator {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Trade> streamSource = env.fromCollection(new ArrayList<Trade>(){
            {
                add(new Trade("12300986754",899,"2018-06"));
                add(new Trade("12300986754",699,"2018-06"));
                add(new Trade("18800986754",88,"2018-07"));
                add(new Trade("18800986754",99,"2018-07"));
                add(new Trade("15800986754",100,"2018-06"));
                add(new Trade("15800986754",200,"2018-06"));
            }
        });

        DataStream<Trade> mapStream = streamSource.keyBy(new KeySelector<Trade, String>() {
            @Override
            public String getKey(Trade trade) throws Exception {
                return trade.getPhone();
            }
        }).reduce(new ReduceFunction<Trade>() {
            @Override
            public Trade reduce(Trade trade01, Trade trade02) throws Exception {
                System.out.println(Thread.currentThread().getName() +"-----" + trade01 + " ----> " + trade02);
                return new Trade(trade01.getPhone(), trade01.getAmount() + trade02.getAmount(), "---------");
            }
        });

        mapStream.print("输出结果");

        // 触发程序执行
        env.execute("reduce stream");
    }
}
